package com.kafka.learn.kafka.first;

import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.util.Collections;
import java.util.Properties;

/**
 * @author 郭辉
 * @className ConsumerListen kafka消费者实现------基于监听器，不需要循环调用
 * @description TODO
 * @date 2022/1/17 11:38
 * @company 海康威视
 */
@Component
public class ConsumerListen {

    /**
     * kafka的IP端口
     * */
    public static final String BROKER_LIST = "localhost:9092";

    /**
     * 订阅的topic
     * */
    public static final String TOPIC = "SNAP_IMAGE_INFO_TOPIC";

    /**
     * 消费者组
     * */
    public static final String GROUP_ID = "group.demo";

    /**
     * KafkaListener
     * */
    @KafkaListener(
            topics = "SNAP_IMAGE_INFO_TOPIC",
            groupId = "group.demo",
            properties = {
            "spring.kafka.bootstrap-servers=localhost:9092"
            }
    )
    public void reciveMessage(ConsumerRecord<?,?> record) {
        System.out.println(record);
        System.out.println(record.value());
    }
}
